package org.raymond.iworks.study.mq.rabbitmq.classic.rpc;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class RPCServer {
    public static final String RPC_QUEUE_NAME = "rpc_queue";

    private static int fib(int n){
        if(n==0) return 0;
        if(n==1) return 1;
        return fib(n-1)+fib(n-2);
    }

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = factory();
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();

        boolean durable = true;
        channel.queueDeclare(RPC_QUEUE_NAME, durable, false, false, null);
        channel.queuePurge(RPC_QUEUE_NAME);

        int prefetchCount = 1;
        channel.basicQos(prefetchCount);

        log.info("[x] Awaiting RPC requests");
        DeliverCallback deliverCallback = (consumerTag, delivery)->{
            AMQP.BasicProperties replyProps = new AMQP.BasicProperties()
                    .builder()
                    // deliveryMode: 消息持久化. MessageProperties.PERSISTENT_TEXT_PLAIN
                    // contentType: mime-type
                    // replyTo: 用于命名一个回调队列,该回调队列用于接收服务端返回的响应报文
                    // correlationId: 在客户端上绑定的回调队列中返回响应报文,需要区分是哪一个请求的,用correlationId来区分,即与具体的某一个请求相关联
                    .correlationId(delivery.getProperties().getCorrelationId())
                    .build();

            String response = "";
            try{
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                int n = Integer.parseInt(message);

                log.info(" [.] fib({})", message);
                response += fib(n);
            }catch (RuntimeException e){
                log.warn(" [.] ", e);
            }finally {
                channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes(StandardCharsets.UTF_8));
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        boolean autoAck = false;
        channel.basicConsume(RPC_QUEUE_NAME, autoAck, deliverCallback, consumerTag->{});
    }

    private static ConnectionFactory factory(){
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.125");
        factory.setPort(5672);
        return factory;
    }
}
